package com.atguigu.state;

import com.atguigu.bean.WaterSensor;
import com.atguigu.functions.MapFunction2Impl;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author gmd
 * @desc 算子状态-广播状态。水位超过指定的阈值发送告警，阈值可以动态修改。
 * @since 2024-12-01 11:34:37
 */
public class OperatorBroadcastState {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // 数据流
        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("127.0.0.1", 7777)
                .map(new MapFunction2Impl());

        // 配置流（用来广播配置）
        DataStreamSource<String> configDS = env.socketTextStream("127.0.0.1", 8888);

        // 1.将 配置流 广播
        MapStateDescriptor<String, Integer> broadcastMapState = new MapStateDescriptor<>("broadcast-state", Types.STRING, Types.INT);
        BroadcastStream<String> configBS = configDS.broadcast(broadcastMapState);

        // 2.把数据流 和 广播后的配置流 connect
        BroadcastConnectedStream<WaterSensor, String> sensorBCS = sensorDS.connect(configBS);

        // 3.调用 process
        sensorBCS.process(
                new BroadcastProcessFunction<WaterSensor, String, String>() {
                    /**
                     * 数据流的处理方法：该方法用于处理数据流，数据流只能读取广播状态，不能修改广播状态。
                     *
                     * @param value   数据流中的元素
                     * @param ctx     ReadOnlyContext对象，提供了对广播状态的只读访问
                     * @param out     Collector对象，用于输出处理结果
                     */
                    @Override
                    public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        // 5.通过上下文获取广播状态，取出里面的值（只读，不能修改）
                        ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(broadcastMapState);
                        Integer threshold = broadcastState.get("threshold");
                        // 判断广播状态里是否有数据，因为刚启动时，可能是数据流的第一条数据先来
                        threshold = (threshold == null ? 0 : threshold);
                        if (value.getVc() > threshold) {
                            out.collect(value + "，水位超过指定的阈值：" + threshold + "!!!");
                        }
                    }

                    /**
                     * 处理广播后的配置流的方法：该方法用于处理广播后的配置流，广播流可以修改广播状态。
                     *
                     * @param value   广播后的配置流中的元素
                     * @param ctx     Context对象，提供了对广播状态的读写访问
                     * @param out     Collector对象，用于输出处理结果
                     */
                    @Override
                    public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                        // 4.通过上下文获取广播状态，往里面写数据
                        BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(broadcastMapState);
                        broadcastState.put("threshold", Integer.valueOf(value));
                    }
                }
        ).print();

        env.execute();
    }

}
